package com.springboot.distributed.transaction.zookeeper;

import java.nio.charset.StandardCharsets;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;
import com.springboot.common.util.StringUtil;
import com.springboot.distributed.transaction.TransactionMessageHandler;
import com.springboot.distributed.transaction.TransactionStatusHandler;

/**
 * zk消息处理器
 *
 * @author supu
 * @date 2019-09-12 11:27
 **/
public class ZookeeperHandler implements TransactionMessageHandler {
    private static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperHandler.class);

    private static final String ROOT_PATH = "/springBootDemo/transaction";
    private static final String SLASH = "/";

    private CuratorFramework curatorFramework;
    private String address;
    private TransactionStatusHandler transactionStatusHandler;

    public ZookeeperHandler(String address) {
        this.address = address;
    }

    @Override
    public void init(TransactionStatusHandler transactionStatusHandler) {
        this.transactionStatusHandler = transactionStatusHandler;
        // 初始化zk客户端
        initZkClient();
        // 开启节点数据监控，处理事务状态消息
        startTransactionMessageWatch();
    }

    private void startTransactionMessageWatch() {
        TreeCache cache = new TreeCache(curatorFramework, ROOT_PATH);
        try {
            cache.getListenable().addListener((curatorFramework, event) -> {
                if (TreeCacheEvent.Type.NODE_ADDED == event.getType()) {
                    ChildData childData = event.getData();
                    if (childData != null) {
                        String message = new String(childData.getData(), StandardCharsets.UTF_8);
                        if (StringUtil.isNotEmpty(message)) {
                            LOGGER.info("收到分布式事务结果消息，path :{}, message:{}", childData.getPath(), message);
                            transactionStatusHandler.handlerTransaction(JSONObject.parseObject(message, TransactionMessage.class));
                        }
                    }
                }
            });
            cache.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void initZkClient() {
        curatorFramework = CuratorFrameworkFactory.newClient(address, new RetryNTimes(10, 10000));
        curatorFramework.start();
    }

    @Override
    public void sendTransactionMessage(TransactionMessage transactionMessage) {
        String transactionMessagePath = getTransactionMessagePath(transactionMessage.getUniqueKey());
        String message = JSONObject.toJSONString(transactionMessage);

        try {
            String result = curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).
                    forPath(transactionMessagePath, message.getBytes(StandardCharsets.UTF_8));
            LOGGER.info("发送分布式事务消息 transactionMessagePath :{}, message:{}, result:{} ", transactionMessagePath, message, result);
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }
    }

    private String getTransactionMessagePath(String key) {
        return ROOT_PATH + SLASH + key + "_" + System.currentTimeMillis();
    }
}
